package com.offcn.bigdata.datastream.transformation

import java.{lang, util}

import org.apache.flink.streaming.api.collector.selector.OutputSelector
import org.apache.flink.streaming.api.scala.StreamExecutionEnvironment
import org.apache.flink.streaming.api.scala._
/**
 * @Auther: BigData-LGW
 * @ClassName: SplitAndSelect
 * @Date: 2020/12/16 20:27
 * @功能描述: $FunctionDescription
 * @Version:1.0
 */
/**
 * 流的拆分split和选择select
 *  datastream使用split操作，将一个流拆分成多个splitstream
 *  我们要向获取其中的某一个流，就是用select来进行选择
 *  基于此我们就可以针对特定的流进行分析处理
 */
object SplitAndSelect {
    def main(args: Array[String]): Unit = {
        val env = StreamExecutionEnvironment.getExecutionEnvironment
        val lines = env.socketTextStream("node-1",9999)
        val goods = lines.map(line => {
            val fields = line.split("\\|")
            val id = fields(0)
            val brand = fields(1)
            val category = fields(2)
            Goods(id,brand,category)
        })
        goods.split(new OutputSelector[Goods] {
            override def select(goods: Goods): lang.Iterable[String] = {
                    goods.category match {
                        case "sports" => {
                            util.Arrays.asList("sports")
                        }
                        case "clothing" => {
                            util.Arrays.asList("clothing")
                        }
                        case "mobile" => {
                            util.Arrays.asList("mobile")
                        }
                    }
                }
            })
            val splitStream = goods.split(goods => Seq(goods.category))
            splitStream.select("clothing").print("clothing:::")
            splitStream.select("sports").print("sports:::")
            env.execute(s"${SplitAndSelect.getClass.getSimpleName}")
    }
}
case class Goods(id: String, brand: String, category: String)